package com.github.yoojia.mqtt.boot;

import com.annimon.stream.Collectors;
import com.annimon.stream.Stream;
import com.annimon.stream.function.Consumer;
import com.annimon.stream.function.Function;
import com.github.yoojia.mqtt.MQTTSocket;
import com.github.yoojia.mqtt.Message;
import com.github.yoojia.mqtt.MessageHandler;
import com.github.yoojia.mqtt.boot.utils.Classes;
import com.github.yoojia.mqtt.boot.utils.ConfigKit;
import com.github.yoojia.mqtt.boot.utils.Keeper;
import com.parkingwang.lang.Lazy;
import com.parkingwang.lang.Supplier;
import com.parkingwang.lang.data.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;

/**
 * @author Yoojia Chen (yoojiachen@gmail.com)
 * @since 1.0.0
 */
public class MQTTBootstrap {

    public static final String VERSION = "NextMQTT(v1.1.0)/Bootstrap";

    private static final Logger LOGGER = LoggerFactory.getLogger(MQTTBootstrap.class);

    private final Lazy<MQTTSocket> mSocket = Lazy.from(new Supplier<MQTTSocket>() {
        @Override
        public MQTTSocket call() {
            final Config broker = mConfig.getConfig("broker");
            final String address = broker.getString("address");
            final String domain = broker.getString("domain");
            final String nodeId = broker.getString("node-id");
            final String username = broker.getString("username");
            final String password = broker.getString("password");
            return MQTTSocket.context()
                    .address(address)
                    .domain(domain)
                    .nodeId(nodeId)
                    .username(username)
                    .password(password)
                    .socket();
        }
    });

    private final Keeper mKeeper = new Keeper(new Runnable() {
        @Override
        public void run() {
            disconnect();
        }
    });

    private final Config mConfig = new Config();
    private final List<ActuatorMeta> mActuatorInfo = new ArrayList<>();
    private final List<Plugin> mPlugins = new ArrayList<>();

    public void start(String fileName) throws Exception {
        final File file = Paths.get(System.getProperty("user.dir"), fileName).toFile();
        if (!file.exists()) {
            ConfigKit.extractByName("/config-sample.yaml", fileName);
        }
        LOGGER.info("Using config file: {}", file.getAbsolutePath());
        try (final InputStream stream = new FileInputStream(file)) {
            mConfig.copyFrom(ConfigKit.stream(stream));
        }
        // 连接服务端
        connect();
    }

    public void join() throws InterruptedException {
        mKeeper.run();
    }

    private void connect() {
        // 连接服务端
        final MQTTSocket socket = mSocket.getChecked();
        socket.connect();
        // 注册组件
        final List<Config> actuatorsConf = mConfig.getConfigList("actuators");
        LOGGER.info("Found Actuators: {}", actuatorsConf.size());
        final List<ActuatorConf> actuators = Stream.of(actuatorsConf)
                .map(new Function<Config, ActuatorConf>() {
                    @Override
                    public ActuatorConf apply(Config config) {
                        return ActuatorConf.of(config);
                    }
                })
                .filter(Predicates.ofActuatorEnabled())
                .collect(Collectors.<ActuatorConf>toList());
        // 注册Req-Rep请求响应类组件
        Stream.of(actuators)
                .filter(Predicates.ofPatternType(PatternType.ReqRep))
                .forEach(new Consumer<ActuatorConf>() {
                    @Override
                    public void accept(ActuatorConf config) {
                        final TagFilter filter = new TagFilter(config.tagFilter);
                        LOGGER.info("Registering <REQ-REP> actuator: {}, Tag-Filter: {}", config.className, config.tagFilter);
                        final ReqRepActuator actuator = config.newReqRepActuator();
                        final long regid = socket.addRequestMessageHandler(filter, new MessageHandler() {
                            @Override
                            public void onMessage(MQTTSocket socket, Message request) {
                                socket.send(actuator.handleRequest(socket, request));
                            }
                        });
                        mActuatorInfo.add(ActuatorMeta.of(regid, config.patternType, actuator, filter));
                        actuator.onInit(socket, config.arguments);
                    }
                });
        // 注册Event事件类组件
        Stream.of(actuators)
                .filter(Predicates.ofPatternType(PatternType.PubSub))
                .forEach(new Consumer<ActuatorConf>() {
                    @Override
                    public void accept(ActuatorConf config) {
                        final TagFilter filter = new TagFilter(config.tagFilter);
                        LOGGER.info("Registering <PUB-SUB> actuator: {}, Tag-Filter: {}", config.className, config.tagFilter);
                        final PubSubActuator actuator = config.newEventActuator();
                        final long regid = socket.addSubMessageHandler(filter, new MessageHandler() {
                            @Override
                            public void onMessage(MQTTSocket socket, Message event) {
                                actuator.onEvent(socket, event);
                            }
                        });
                        mActuatorInfo.add(ActuatorMeta.of(regid, config.patternType, actuator, filter));
                        actuator.onInit(socket, config.arguments);
                    }
                });
        // 注册功能插件类组件
        final List<Config> pluginConf = mConfig.getConfigList("plugins");
        LOGGER.info("Found Plugins: {}", pluginConf.size());
        Stream.of(pluginConf)
                .map(new Function<Config, PluginConf>() {
                    @Override
                    public PluginConf apply(Config config) {
                        return PluginConf.of(config);
                    }
                })
                .filter(Predicates.ofPluginEnabled())
                .forEach(new Consumer<PluginConf>() {
                    @Override
                    public void accept(PluginConf config) {
                        LOGGER.info("Registering <Plugin> actuator: {}", config.className);
                        final Plugin plugin = Classes.newInstance(config.className);
                        mPlugins.add(plugin);
                        plugin.onInit(socket, config.arguments);
                        plugin.onStart();
                    }
                });
        LOGGER.info("Start...");
    }

    private void disconnect() {
        // 停止插件组件
        Stream.of(mPlugins)
                .forEach(new Consumer<Plugin>() {
                    @Override
                    public void accept(Plugin plugin) {
                        plugin.onStop();
                    }
                });
        // 移除消息处理器
        mSocket.ifPresent(new com.parkingwang.lang.Consumer<MQTTSocket>() {
            @Override
            public void call(MQTTSocket socket) {
                for (ActuatorMeta meta : mActuatorInfo) {
                    if (PatternType.ReqRep.equals(meta.patternType)) {
                        socket.removeRequestMessageHandler(meta.regid);
                    } else if (PatternType.PubSub.equals(meta.patternType)) {
                        socket.removeSubMessageHandler(meta.regid);
                    }
                }
                socket.disconnect();
            }
        });
        mPlugins.clear();
        mActuatorInfo.clear();
        LOGGER.info("Stop...");
    }
}
